//package com.atguigu.app.dws;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.atguigu.bean.TradeProvinceOrderWindow;
//import com.atguigu.func.DimAsyncJoinFunction;
//import com.atguigu.utils.DateFormatUtil;
//import com.atguigu.utils.KafkaUtil;
//import com.atguigu.utils.MyClickHouseUtil;
//import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.FlatMapFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.common.functions.ReduceFunction;
//import org.apache.flink.api.common.functions.RichFilterFunction;
//import org.apache.flink.api.common.state.StateTtlConfig;
//import org.apache.flink.api.common.state.ValueState;
//import org.apache.flink.api.common.state.ValueStateDescriptor;
//import org.apache.flink.api.common.time.Time;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.AsyncDataStream;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//import org.apache.flink.util.Collector;
//
//import java.time.Duration;
//import java.util.HashSet;
//import java.util.concurrent.TimeUnit;
////方案三：left join产生的重复数据问题：下游需求是累加类指标，那么后到的数据将相关字段置为0写出
//
//
////数据流：web/app -> mysql -> maxwell -> kafka(ods) -> flinkApp(订单明细关联后的事实表) -> kafka(dwd) ->flinkApp -> Clickhouse(DWS)
////程 序：Mock      ->mysql -> maxwell -> kafka(zk) -> DwdTradeOrderDetail ->kafka(zk) ->Dws10TradeProvinceOrderWindow(redis,hdfs,zk,hbase,phoenix)->clickhouse(zk)
///*
//10.10 交易域省份粒度下单各窗口汇总表
//（按照事件时间开窗，一个订单会有多个订单明细，同一个订单的订单明细的事件时间肯定会一样，会进到同一个窗口，所以我们只需要对同一个窗口对order_id进行去重即可）
//10.10.1 主要任务
//从 Kafka 读取业务数据，筛选订单表数据，统计各省份各窗口订单数和订单金额，将数据写入 ClickHouse 交易域省份粒度下单各窗口汇总表。
//
// */
////todo 1.获取执行环境
////todo 2.读取dwd层订单明细主题创建流（dwd层订单明细时通过left join产生的，会产生左 null，null，左右）
////todo 3.过滤null值并转化为json对象
////todo 4.按照订单明细id分组，去重数据
////todo 5.将数据转换为JavaBean对象
////todo 6.提取时间戳生成watermark
////todo 7.按照省份id分组进行开窗聚合
////todo 8.关联省份表补充维度信息(聚合后再关联，数据量就会变少很多)
////todo 9.将数据写到clickhouse
////todo 10.启动任务
//public class Dws10TradeProvinceOrderWindow03 {
//    public static void main(String[] args) throws Exception {
//        //todo 1.获取执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        //todo 生产环境一定要写，测试注释掉，否则每次测试都得开hdfs
////        需要从checkpoint或savepoint启动程序
////        //2.1 开启checkpoint，每隔5s钟做一次ck，并指定ck的一致性语义
////        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);//exactly once：默认barrier对齐
////        //2.2 设置超时时间为1min
////        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);//设置超时时间设置checkpoint的超时时间为1min，是指做一次checkpoint的时间；如果超时则认为本次checkpoint失败，这个checkpoint就丢了，继续一下一次checkpoint即可
////        //2.3设置两次重启的最小时间间隔为3s
////        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
////        //2.4设置任务关闭的时候保留最后一次ck数据
////        env.getCheckpointConfig().enableExternalizedCheckpoints(
////                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
////        );
////        //2.5 指定从ck自动重启策略
////        env.setRestartStrategy(RestartStrategies.failureRateRestart(
////                3, Time.days(1L),Time.minutes(1L)
////        ));
////        //2.6 设置状态后端
////        env.setStateBackend(new HashMapStateBackend());//本地状态位置
////        env.getCheckpointConfig().setCheckpointStorage(
////                "hdfs://hadoop102:8020/flinkCDC/220828"
////        );//checkpoint状态位置
////        //2.7 设置访问HDFS的用户名
////        System.setProperty("HADOOP_USER_NAME","atguigu");
//
//        //todo 2.读取dwd层订单明细主题创建流（dwd层订单明细时通过left join产生的，会产生左 null，null，左右）
//        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", "province_order1_220828").setStartFromEarliest());
//
//        //todo 3.过滤null值并转化为json对象
//        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
//            @Override
//            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
//                if (value != null) {
//                    JSONObject jsonObject = JSON.parseObject(value);
//                    out.collect(jsonObject);
//                }
//            }
//        });
////        jsonObjDS.print("jsonObjDS>>>>>>");
//
//        //todo 4.按照订单明细id分组，去重数据(主键本来唯一，但是由于left join产生多个)，并转化为javabean
//        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.keyBy(json -> json.getString("id"))
//
////        filterDS.print("filterDS>>>>");
//
//        //todo 5.将数据转换为JavaBean对象
//        SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceOrderDS = filterDS.map(new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
//            @Override
//            public TradeProvinceOrderWindow map(JSONObject value) throws Exception {
//
//                HashSet<String> orderIds = new HashSet<>();
//                orderIds.add(value.getString("order_id"));
//
//                return TradeProvinceOrderWindow.builder()
//                        .provinceId(value.getString("province_id"))
//                        .orderIdSet(orderIds)//多个订单明细id（id不相同）的order_id相同，order_id进到Set集合里会自动去重，对应的事件时间是一样的，会进到一个窗口
//                        .orderAmount(value.getBigDecimal("split_total_amount"))//原始金额，一定是有值的，不用判断是否为null
////                        .ts(value.getLong("create_time"))
//                        .ts(DateFormatUtil.toTs(value.getString("create_time"),true))
//                        .build();
//            }
//        });
//
////        provinceOrderDS.print("provinceOrderDS>>>>");
//        //todo 6.提取时间戳生成watermark
//        SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceOrderWithWMDS = provinceOrderDS.assignTimestampsAndWatermarks(
//                WatermarkStrategy
//                        .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2))
//                        .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindow>() {
//                            @Override
//                            public long extractTimestamp(TradeProvinceOrderWindow element, long recordTimestamp) {
//                                return element.getTs();
//                            }
//                        })
//        );
//
//        //todo 7.按照省份id分组进行开窗聚合
//        SingleOutputStreamOperator<TradeProvinceOrderWindow> reducedDS = provinceOrderWithWMDS.keyBy(s -> s.getProvinceId())
//                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
//                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
//                    @Override
//                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
//
//                        //将窗口里来的一条条javabean里的oderSet里的数据聚合成一个set
//                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());//将v2也就是后面数据的orderSet里的订单id添加进第一条数据的订单id里，并去重了，去重后的订单id都不相同
//                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
//                        return value1;
//
//                    }
//                }, new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
//                    @Override
//                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
//                        //获取数据
//                        TradeProvinceOrderWindow next = input.iterator().next();
//
//                        //todo 补充去重后的订单数(也就是开窗聚合后的唯一的数据的orderSet的size大小)
//                        next.setOrderCount((long) next.getOrderIdSet().size());
//                        //补充信息
//                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
//                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
//                        next.setTs(System.currentTimeMillis());
//
//                        //一定不能忘了要把数据收集发送给下游
//                        out.collect(next);
//
//
//                    }
//                });
//
////        provinceOrderWithWMDS.print("provinceOrderWithWMDS>>>>>>");//有数据
//        reducedDS.print("reducedDS>>>>>>");//没有数据
//
//        //todo 8.关联省份表补充维度信息(聚合后再关联，数据量就会变少很多)
//        SingleOutputStreamOperator<TradeProvinceOrderWindow> resultDS = AsyncDataStream.unorderedWait(reducedDS, new DimAsyncJoinFunction<TradeProvinceOrderWindow>("DIM_BASE_PROVINCE") {
//            @Override
//            public String getKey(TradeProvinceOrderWindow input) {
//                return input.getProvinceId();//获取上游设置进去的省份id用来关联省份维表
//            }
//
//            @Override
//            public void join(TradeProvinceOrderWindow input, JSONObject dimInfo) {
//                input.setProvinceName(dimInfo.getString("NAME"));//获取phoenix查询出来的省份名，要看sink_columns的字段
//
//            }
//        }, 60, TimeUnit.SECONDS);
//
//        resultDS.print("即将写入clickhouse的数据,跳过oderIdSet");
//
//        //todo 9.将数据写到clickhouse
//        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)"));
//
//        //todo 10.启动任务
//        env.execute("Dws10TradeProvinceOrderWindow");
//
//    }
//}
